package com.nulldev.util.concurrency.threadIt.v4.emapi.cft;

import static com.nulldev.util.concurrency.threadIt.v4.emapi.uft.ExecutableUtils.callAsyncEventsOrFail;
import static com.nulldev.util.concurrency.threadIt.v4.emapi.uft.ExecutableUtils.callAsyncFails;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import com.nulldev.util.JVM.Arguments;
import com.nulldev.util.concurrency.threadIt.v3.extensions.TaskStatus;
import com.nulldev.util.concurrency.threadIt.v4.asyncTasks.AsyncTaskCallback;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.data.Arrays.queues.EnhancedLinkedBlockingQueue;
import com.nulldev.util.internal.backport.concurrency9.Lists;
import com.nulldev.util.internal.backport.concurrency9.concurrent.CompletableFuture;

public abstract class CompleteableFutureTask<T> implements Runnable, Executable<T> {

	/*
	 * Since we have a backport of JDK's CompletableFuture<T> and it's [it was] a
	 * tiny bit faster, why the fuck not?
	 */

	private final static boolean ENABLE_OFFLOADED_ASYNC = Arguments.notArguments("--Tasks.disableAsyncOffload", "--CompleteableFutureTask.disableAsyncOffload");
	private final static boolean ENABLE_ELBQ = Arguments.hasArgument("--Tasks.enableELBQ");
	private final static boolean USE_NEW_FAIL_LOGIC = Arguments.notArguments("--Tasks.disableNewFailureLogic");

	private final BlockingQueue<AsyncTaskCallback<T>> callback;
	private final AtomicBoolean completed = new AtomicBoolean();
	private final CompletableFuture<T> o;
	private final AtomicReference<T> result = new AtomicReference<T>();
	private final AtomicReference<TaskStatus> status = new AtomicReference<TaskStatus>(TaskStatus.QUEUED);

	public CompleteableFutureTask() {
		this.o = new CompletableFuture<T>();
		this.callback = ENABLE_ELBQ ? new EnhancedLinkedBlockingQueue<AsyncTaskCallback<T>>() : new LinkedTransferQueue<AsyncTaskCallback<T>>();
	}

	public CompleteableFutureTask(final BlockingQueue<AsyncTaskCallback<T>> callback) {
		this.o = new CompletableFuture<T>();
		this.callback = callback;
	}

	@Override
	public T run_st() {
		this.run();
		return this.result.get();
	}

	@Override
	public void run() {
		this.status.lazySet(TaskStatus.RUNNING);
		try {
			final T _result = this.execute();
			this.completed.lazySet(true);
			this.status.lazySet(TaskStatus.COMPLETED);
			this.result.set(_result);
			this.o.complete(_result);

			callAsyncEventsOrFail(ENABLE_OFFLOADED_ASYNC, false, this.callback, _result);
		} catch (Throwable e) {
			this.o.cancel(true);
			this.status.lazySet(TaskStatus.FAILED);

			callAsyncFails(ENABLE_OFFLOADED_ASYNC, false, this.callback, e);
			if (!USE_NEW_FAIL_LOGIC)
				throw new RuntimeException(e);
		}
	}

	@Override
	public List<AsyncTaskCallback<?>> callbacks() {
		return Lists.of(this.callback.toArray(new AsyncTaskCallback[this.callback.size()]));
	}

	public abstract T execute();

	@Override
	public T waitTillCompletion() {
		if (this.completed.get()) {
			return this.result.get();
		} else if (this.o.isCancelled()) {
			return this.result.get();
		} else {
			return this.o.join();
		}
	}

	@Override
	public boolean addCallbackTask(final AsyncTaskCallback<T> asyncTaskCallback) {
		if (asyncTaskCallback == null) {
			return false;
		} else if (this.completed()) {

			if (this.o.isCompletedExceptionally()) {
				asyncTaskCallback.onError(new RuntimeException());
			} else {
				asyncTaskCallback.onCompletion(result.get());
			}
			return false;
		} else {
			try {
				return this.callback.offer(asyncTaskCallback);
			} catch (Exception e) {
				if (this.completed()) {
					if (this.o.isCompletedExceptionally()) {
						asyncTaskCallback.onError(new RuntimeException());
					} else {
						asyncTaskCallback.onCompletion(result.get());
					}
				}
				return false;
			}
		}
	}

	@Override
	public T result() {
		return this.result.get();
	}

	public boolean completed() {
		return this.completed.get();
	}

	public TaskStatus status() {
		return this.status.get();
	}

	@Override
	public boolean hasTrace() {
		return false;
	}

	@Override
	public void asyncSignalQueued() {
		// NO-OP
	}

	@Override
	public void signalQueued() {
		// NO-OP
	}

	@Override
	public void rejectSubmission() {
		// NO-OP
	}

	@Override
	public int getFlags() {
		return 0;
	}

	@Override
	public Object getFlag(final int flag) {
		return FT_NOT_SUPPORTED;
	}

	@Override
	public T waitFor(final long amount, final TimeUnit unit) {
		if (unit == null)
			throw new NullPointerException("unit == null");

		if (amount <= 0)
			throw new IllegalArgumentException("amount <= 0!");

		if (this.completed())
			return this.result.get();
		try {
			return this.o.get(amount, unit);
		} catch (Exception e) {
			return this.o.getNow(this.result.get());
		}
	}

	/**
	 * Attempts to kill the executing thread.
	 */
	public void kill() {
		this.o.cancel(true);
		this.status.lazySet(TaskStatus.KILLED);
		this.completed.lazySet(true);
//		if (this.t != null) {
//			this.t.interrupt();
//		}else {
//			Thread.currentThread().interrupt();
//		}
	}
}
